package com.jacquesh;

import com.jacquesh.model.CoffeeShop;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

public class FlinkDemo0003_1 {
    public static void main(String[] args) throws Exception {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        StreamExecutionEnvironment environment=
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream= environment
                .socketTextStream("192.168.88.180",8888);
     DataStream <CoffeeShop> dataStream=   stream.flatMap(new RichFlatMapFunction<String, CoffeeShop>() {
         @Override
         public void flatMap(String s, Collector<CoffeeShop> collector) {
             try{
                 String [] sp= s.split(",");
                 collector.collect(CoffeeShop.builder()
                         .id(Long.parseLong(sp[0]))
                         .date(format.parse(sp[1]))
                         .sales(Integer.parseInt(sp[2]))
                         .build());
             }catch (Exception ex){
                 System.out.println(ex);
             }

         }
     })        /** 依据id分组*/
                .keyBy(CoffeeShop::getId)
                 /** 对销量进行求和*/
                .sum("sales");
        dataStream.print().setParallelism(1);
        environment.execute("keyBy &  sum demo");
    }
}
